use std::time::Duration;

use async_std::net::TcpStream;
use futures::{SinkExt, StreamExt};

use bytes::Bytes;
use honeybee::prelude::*;
use honeybee::Result;
use honeybee_proto::hall;
use prost::Message as ProstMessage;

use crate::mongo::account;

use super::message_process;
use super::token;
use super::user::{Disconnected, User};

pub struct UserStream {
    addr: Option<Addr<Self>>,
    user_addr: Option<Addr<User>>,
    stream: TcpStream,
    peer_addr: String,
    framed_write: FramedWrite<Compat<TcpStream>, UserMsgCodec>,
}

impl UserStream {
    pub fn new(stream: TcpStream) -> Self {
        if let Err(err) = stream.set_nodelay(true) {
            error!("user stream {:?} set_nodelay err:{}", stream, err);
        }
        let framed_write = FramedWrite::new(stream.clone().compat(), UserMsgCodec::new());
        let peer_addr = stream
            .peer_addr()
            .map(|addr| addr.to_string())
            .unwrap_or_else(|_| "None".into());
        Self {
            addr: None,
            user_addr: None,
            stream,
            peer_addr,
            framed_write,
        }
    }

    fn peer_addr(&self) -> &String {
        &self.peer_addr
    }

    async fn send<T: 'static + MessageId + prost::Message>(&mut self, msg: &T) -> Result<()> {
        self.framed_write.send(msg).await
    }
}

#[async_trait::async_trait]
impl Actor for UserStream {
    fn name(&self) -> &str {
        "UserStream"
    }

    async fn started(&mut self, ctx: &mut Context<Self>) -> Result<()> {
        let addr = ctx.address();
        addr.send(Login).ok();
        self.addr = Some(addr);
        info!("{} actor:{} started", self.name(), ctx.actor_id());
        Ok(())
    }

    async fn stopped(&mut self, ctx: &mut Context<Self>) {
        if let Some(addr) = self.user_addr.take() {
            let _ = addr.send(Disconnected);
        }
        info!(
            "{} actor:{} stoped, {} disconnect",
            self.name(),
            ctx.actor_id(),
            self.peer_addr()
        );
    }
}

#[message]
struct Login;

#[async_trait::async_trait]
impl Handler<Login> for UserStream {
    async fn handle(&mut self, ctx: &mut Context<Self>, _: Login) -> Result<()> {
        let addr = self.addr.take().ok_or_else(|| anyhow!("inner error"))?;

        let addr_ = addr.clone();
        let mut defer = Defer::new(move || {
            let _ = addr_.stop(None);
        });

        let mut framed_read = FramedRead::new(self.stream.clone().compat(), UserMsgCodec::new());

        let login_msg = timeout(Duration::from_secs(1), framed_read.next())
            .await
            .map_err(|err| anyhow!("login timeout:{}", err))?
            .ok_or_else(|| anyhow!("login close by peer"))?
            .map_err(|err| anyhow!("login receive failed, err:{}", err))?;

        let login = hall::ReqLogin::decode(login_msg.bytes)
            .map_err(|err| anyhow!("login failed, err:{}", err))?;

        debug!("receive login token:{}", login.token);

        let account = token::request_account(&login.token).await?;
        let account = account::get_or_create(&account).await?;

        self.user_addr = Some(
            User::new(account, addr, self.peer_addr.clone())
                .await?
                .start()
                .await?,
        );

        ctx.add_stream(framed_read);

        //defer.cancel 放在最后,防止上面异常导致未stop
        defer.cancel();

        Ok(())
    }
}

type MsgResult = std::result::Result<UserRawMsg, anyhow::Error>;
#[async_trait::async_trait]
impl StreamHandler<MsgResult> for UserStream {
    async fn handle(&mut self, ctx: &mut Context<Self>, msg: MsgResult) {
        match msg {
            Ok(msg) => {
                let msg_id = msg.id;
                if let Some(ref user_addr) = self.user_addr {
                    if let Err(err) = message_process(user_addr, msg) {
                        error!("{} process msg:{} error:{}", self.name(), msg_id, err)
                    }
                } else {
                    error!(
                        "{} process msg:{} error, user_addr is None",
                        self.name(),
                        msg_id
                    )
                }
            }
            Err(err) => {
                error!("{} receive message error:{}", self.name(), err);
                ctx.stop(None);
            }
        }
    }

    async fn finished(&mut self, ctx: &mut Context<Self>) {
        error!("{} disconnected", self.name());
        let _ = ctx.stop(None);
    }
}

pub struct SendUserMsg<T: 'static + MessageId + prost::Message>(pub T);

impl<T: 'static + MessageId + prost::Message> Message for SendUserMsg<T> {
    type Result = ();
}

#[async_trait::async_trait]
impl<T: 'static + MessageId + prost::Message> Handler<SendUserMsg<T>> for UserStream {
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: SendUserMsg<T>) -> Result<()> {
        if let Err(err) = self.send(&msg.0).await {
            error!("{} send msg err:{}", self.name(), err);
            return Ok(());
        }
        Ok(())
    }
}

#[message]
pub struct UserMsgForward(pub u16, pub Vec<u8>);

#[async_trait::async_trait]
impl Handler<UserMsgForward> for UserStream {
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: UserMsgForward) -> Result<()> {
        if let Err(err) = self
            .framed_write
            .send(UserRawMsg {
                id: msg.0,
                bytes: Bytes::from(msg.1),
            })
            .await
        {
            error!(
                "{} msg:{} forward send to user failed:{}",
                self.name(),
                msg.0,
                err
            );
        }
        Ok(())
    }
}
